Support service framework to separate plugin logic

This fix supports service framework in order to separate
the plugin code into backend specific and generic part.

Change-Id: I66d0f4a9ce5453ec7ca0bf0fe52f24a3a2e3c8ea
Closes-Bug: #1567770

devstack/README.rst

@@ -8,3 +8,6 @@ A `local.conf` recipe to enable tap-as-a-service::
     enable_plugin tap-as-a-service https://github.com/openstack/tap-as-a-service
     enable_service taas
     enable_service taas_openvswitch_agent
+    Q_PLUGIN_EXTRA_CONF_PATH=/etc/neutron
+    Q_PLUGIN_EXTRA_CONF_FILES=(taas_plugin.ini)
+    TAAS_SERVICE_DRIVER=TAAS:TAAS:neutron_taas.services.taas.service_drivers.taas_rpc.TaasRpcDriver:default

devstack/devstackgaterc

@@ -27,6 +27,10 @@ OVERRIDE_ENABLED_SERVICES+=,taas,taas_openvswitch_agent
 OVERRIDE_ENABLED_SERVICES+=,tempest,dstat
 export OVERRIDE_ENABLED_SERVICES
 
+# Set config file
+export DEVSTACK_LOCAL_CONFIG+=$'\n'"Q_PLUGIN_EXTRA_CONF_PATH=/etc/neutron"
+export DEVSTACK_LOCAL_CONFIG+=$'\n'"Q_PLUGIN_EXTRA_CONF_FILES=(taas_plugin.ini)"
+
 # Begin list of exclusions.
 r="^(?!.*"
 

devstack/plugin.sh

@@ -17,24 +17,23 @@
 # This script is meant to be sourced from devstack.  It is a wrapper of
 # devmido scripts that allows proper exporting of environment variables.
 
-ABSOLUTE_PATH=$(cd `dirname "${BASH_SOURCE[0]}"` && pwd)
-PLUGIN_PATH=$ABSOLUTE_PATH/..
-
-TAAS_OVS_AGENT_BINARY="$NEUTRON_BIN_DIR/neutron-taas-openvswitch-agent"
-TAAS_OVS_AGENT_CONF_FILE="/etc/neutron/taas.ini"
 
 function install_taas {
-    pip_install --no-deps --editable $PLUGIN_PATH
+    pip_install --no-deps --editable $TAAS_PLUGIN_PATH
 }
 
 function configure_taas_plugin {
+    if [ ! -d $NEUTRON_CONF_DIR ]; then
+        _create_neutron_conf_dir
+    fi
+    cp $TAAS_PLUGIN_PATH/etc/taas_plugin.ini $TAAS_PLUGIN_CONF_FILE
     _neutron_service_plugin_class_add taas
 }
 
 function configure_taas_openvswitch_agent {
     local conf=$TAAS_OVS_AGENT_CONF_FILE
 
-    cp $PLUGIN_PATH/etc/taas.ini $conf
+    cp $TAAS_PLUGIN_PATH/etc/taas.ini $conf
     iniset $conf taas driver neutron_taas.services.taas.drivers.linux.ovs_taas.OvsTaasDriver
     iniset $conf taas enabled True
     iniset $conf taas vlan_range_start 3000
@@ -54,6 +53,11 @@ if is_service_enabled taas; then
             configure_taas_plugin
         elif [[ "$2" == "post-config" ]]; then
             neutron-db-manage --subproject tap-as-a-service upgrade head
+            echo "Configuring taas"
+            if [ "$TAAS_SERVICE_DRIVER" ]; then
+                inicomment $TAAS_PLUGIN_CONF_FILE service_providers service_provider
+                iniadd $TAAS_PLUGIN_CONF_FILE service_providers service_provider $TAAS_SERVICE_DRIVER
+            fi
         elif [[ "$2" == "extra" ]]; then
             :
         fi

devstack/settings

@@ -0,0 +1,6 @@
+# Devstack settings
+ABSOLUTE_PATH=$(cd `dirname "${BASH_SOURCE[0]}"` && pwd)
+TAAS_PLUGIN_PATH=$ABSOLUTE_PATH/..
+TAAS_PLUGIN_CONF_FILE="/etc/neutron/taas_plugin.ini"
+TAAS_OVS_AGENT_BINARY="$NEUTRON_BIN_DIR/neutron-taas-openvswitch-agent"
+TAAS_OVS_AGENT_CONF_FILE="/etc/neutron/taas.ini"

etc/taas_plugin.ini

@@ -0,0 +1,7 @@
+[DEFAULT]
+
+
+[service_providers]
+# Defines providers for advanced services using the format:
+# <service_type>:<name>:<driver>[:default] (multi valued)
+service_provider = TAAS:TAAS:neutron_taas.services.taas.service_drivers.taas_rpc.TaasRpcDriver:default

neutron_taas/db/migration/alembic_migration/versions/CONTRACT_HEAD

@@ -1 +1 @@
-1817af933379
+2ecce0368a62

neutron_taas/db/migration/alembic_migration/versions/newton/contract/2ecce0368a62_add_foreign_key_constraint_on_tap_id_association.py

@@ -0,0 +1,37 @@
+# Copyright 2016 Midokura SARL
+#
+# 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.
+
+"""add foreign key constraint on tap id association
+
+Revision ID: 2ecce0368a62
+Revises: 1817af933379
+Create Date: 2016-05-19 11:39:52.892610
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '2ecce0368a62'
+down_revision = '1817af933379'
+
+from alembic import op
+
+
+def upgrade():
+    op.create_foreign_key(
+        constraint_name=None,
+        source_table='tap_id_associations',
+        referent_table='tap_services',
+        local_cols=['tap_service_id'],
+        remote_cols=['id'],
+        ondelete='CASCADE')

neutron_taas/db/taas_db.py

@@ -22,6 +22,7 @@ from neutron_taas.extensions import taas
 from oslo_log import log as logging
 from oslo_utils import uuidutils
 import sqlalchemy as sa
+from sqlalchemy import orm
 from sqlalchemy.orm import exc
 
 
@@ -57,8 +58,15 @@ class TapIdAssociation(model_base.BASEV2):
     # Internal mapping between a TAP Service and
     # id to be used by the Agents
     __tablename__ = 'tap_id_associations'
-    tap_service_id = sa.Column(sa.String(36))
+    tap_service_id = sa.Column(sa.String(36),
+                               sa.ForeignKey("tap_services.id",
+                                             ondelete='CASCADE'))
     taas_id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
+    tap_service = orm.relationship(
+        TapService,
+        backref=orm.backref("tap_service_id",
+                            lazy="joined", cascade="delete"),
+        primaryjoin='TapService.id==TapIdAssociation.tap_service_id')
 
 
 class Tass_db_Mixin(taas.TaasPluginBase, base_db.CommonDbMixin):
@@ -126,14 +134,18 @@ class Tass_db_Mixin(taas.TaasPluginBase, base_db.CommonDbMixin):
             )
             context.session.add(tap_service_db)
 
+        return self._make_tap_service_dict(tap_service_db)
+
+    def create_tap_id_association(self, context, tap_service_id):
+        LOG.debug("create_tap_id_association() called")
         # create the TapIdAssociation object
         with context.session.begin(subtransactions=True):
             tap_id_association_db = TapIdAssociation(
-                tap_service_id=tap_service_db['id']
+                tap_service_id=tap_service_id
                 )
             context.session.add(tap_id_association_db)
 
-        return self._make_tap_service_dict(tap_service_db)
+        return self._make_tap_id_association_dict(tap_id_association_db)
 
     def create_tap_flow(self, context, tap_flow):
         LOG.debug("create_tap_flow() called")
@@ -163,9 +175,6 @@ class Tass_db_Mixin(taas.TaasPluginBase, base_db.CommonDbMixin):
         if not count:
             raise taas.TapServiceNotFound(tap_id=id)
 
-        context.session.query(TapIdAssociation).filter_by(
-            tap_service_id=id).delete()
-
     def delete_tap_flow(self, context, id):
         LOG.debug("delete_tap_flow() called")
 

neutron_taas/services/taas/service_drivers/__init__.py

@@ -0,0 +1,63 @@
+# Copyright (C) 2016 Midokura SARL.
+#
+# 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 abc
+
+from oslo_log import log as logging
+import six
+
+LOG = logging.getLogger(__name__)
+
+
+@six.add_metaclass(abc.ABCMeta)
+class TaasBaseDriver(object):
+
+    def __init__(self, service_plugin):
+        self.service_plugin = service_plugin
+
+    @property
+    def service_type(self):
+        pass
+
+    @abc.abstractmethod
+    def create_tap_service_precommit(self, context):
+        pass
+
+    @abc.abstractmethod
+    def delete_tap_service_precommit(self, context):
+        pass
+
+    @abc.abstractmethod
+    def create_tap_flow_precommit(self, context):
+        pass
+
+    @abc.abstractmethod
+    def delete_tap_flow_precommit(self, context):
+        pass
+
+    @abc.abstractmethod
+    def create_tap_service_postcommit(self, context):
+        pass
+
+    @abc.abstractmethod
+    def delete_tap_service_postcommit(self, context):
+        pass
+
+    @abc.abstractmethod
+    def create_tap_flow_postcommit(self, context):
+        pass
+
+    @abc.abstractmethod
+    def delete_tap_flow_postcommit(self, context):
+        pass

neutron_taas/services/taas/service_drivers/service_driver_context.py

@@ -0,0 +1,67 @@
+# Copyright (C) 2016 Midokura SARL.
+#
+# 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_taas.extensions import taas
+
+from oslo_log import log
+
+LOG = log.getLogger(__name__)
+
+
+class ServiceDriverContext(object):
+    """ServiceDriverContext context base class"""
+    def __init__(self, service_plugin, plugin_context):
+        self._plugin = service_plugin
+        self._plugin_context = plugin_context
+
+
+class TapServiceContext(ServiceDriverContext):
+
+    def __init__(self, service_plugin, plugin_context, tap_service):
+        super(TapServiceContext, self).__init__(service_plugin, plugin_context)
+        self._tap_service = tap_service
+        self._tap_id_association = None
+        self._setup_tap_id_association(tap_service['id'])
+
+    def _setup_tap_id_association(self, tap_service_id):
+        try:
+            self._tap_id_association = self._plugin.get_tap_id_association(
+                self._plugin_context, tap_service_id)
+        except taas.TapServiceNotFound:
+            LOG.debug("Not found tap_ip_association for tap_service: %s",
+                      tap_service_id)
+
+    @property
+    def tap_service(self):
+        return self._tap_service
+
+    @property
+    def tap_id_association(self):
+        return self._tap_id_association
+
+    @tap_id_association.setter
+    def tap_id_association(self, tap_id_association):
+        """Set tap_id_association in context"""
+        self._tap_id_association = tap_id_association
+
+
+class TapFlowContext(ServiceDriverContext):
+
+    def __init__(self, service_plugin, plugin_context, tap_flow):
+        super(TapFlowContext, self).__init__(service_plugin, plugin_context)
+        self._tap_flow = tap_flow
+
+    @property
+    def tap_flow(self):
+        return self._tap_flow

neutron_taas/services/taas/service_drivers/taas_agent_api.py

@@ -0,0 +1,81 @@
+# Copyright (C) 2016 Midokura SARL.
+# Copyright (C) 2015 Ericsson AB
+# Copyright (c) 2015 Gigamon
+#
+# 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.common import rpc as n_rpc
+
+from oslo_log import log as logging
+import oslo_messaging as messaging
+
+LOG = logging.getLogger(__name__)
+
+
+class TaasCallbacks(object):
+    """Currently there are no callbacks to the Taas Plugin."""
+
+    def __init__(self, plugin):
+        super(TaasCallbacks, self).__init__()
+        self.plugin = plugin
+        return
+
+
+class TaasAgentApi(object):
+    """RPC calls to agent APIs"""
+
+    def __init__(self, topic, host):
+        self.host = host
+        target = messaging.Target(topic=topic, version='1.0')
+        self.client = n_rpc.get_client(target)
+        return
+
+    def create_tap_service(self, context, tap_service, host):
+        LOG.debug("In RPC Call for Create Tap Service: Host=%s, MSG=%s" %
+                  (host, tap_service))
+
+        cctxt = self.client.prepare(fanout=True)
+        cctxt.cast(context, 'create_tap_service', tap_service=tap_service,
+                   host=host)
+
+        return
+
+    def create_tap_flow(self, context, tap_flow_msg, host):
+        LOG.debug("In RPC Call for Create Tap Flow: Host=%s, MSG=%s" %
+                  (host, tap_flow_msg))
+
+        cctxt = self.client.prepare(fanout=True)
+        cctxt.cast(context, 'create_tap_flow', tap_flow_msg=tap_flow_msg,
+                   host=host)
+
+        return
+
+    def delete_tap_service(self, context, tap_service, host):
+        LOG.debug("In RPC Call for Delete Tap Service: Host=%s, MSG=%s" %
+                  (host, tap_service))
+
+        cctxt = self.client.prepare(fanout=True)
+        cctxt.cast(context, 'delete_tap_service', tap_service=tap_service,
+                   host=host)
+
+        return
+
+    def delete_tap_flow(self, context, tap_flow_msg, host):
+        LOG.debug("In RPC Call for Delete Tap Flow: Host=%s, MSG=%s" %
+                  (host, tap_flow_msg))
+
+        cctxt = self.client.prepare(fanout=True)
+        cctxt.cast(context, 'delete_tap_flow', tap_flow_msg=tap_flow_msg,
+                   host=host)
+
+        return

neutron_taas/services/taas/service_drivers/taas_rpc.py

@@ -0,0 +1,157 @@
+# Copyright (C) 2016 Midokura SARL.
+# Copyright (C) 2015 Ericsson AB
+# Copyright (c) 2015 Gigamon
+#
+# 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.common import rpc as n_rpc
+from neutron_taas.common import topics
+from neutron_taas.extensions import taas as taas_ex
+from neutron_taas.services.taas import service_drivers
+from neutron_taas.services.taas.service_drivers import taas_agent_api
+
+from oslo_config import cfg
+from oslo_log import log as logging
+
+LOG = logging.getLogger(__name__)
+
+
+class TaasRpcDriver(service_drivers.TaasBaseDriver):
+    """Taas Rpc Service Driver class"""
+
+    def __init__(self, service_plugin):
+        LOG.debug("Loading TaasRpcDriver.")
+        super(TaasRpcDriver, self).__init__(service_plugin)
+        self.endpoints = [taas_agent_api.TaasCallbacks(service_plugin)]
+        self.conn = n_rpc.create_connection()
+        self.conn.create_consumer(topics.TAAS_PLUGIN,
+                                  self.endpoints, fanout=False)
+
+        self.conn.consume_in_threads()
+
+        self.agent_rpc = taas_agent_api.TaasAgentApi(
+            topics.TAAS_AGENT,
+            cfg.CONF.host
+        )
+
+        return
+
+    def _get_taas_id(self, context, tf):
+        ts = self.service_plugin.get_tap_service(context,
+                                                 tf['tap_service_id'])
+        taas_id = (self.service_plugin.get_tap_id_association(
+            context,
+            tap_service_id=ts['id'])['taas_id'] +
+            cfg.CONF.taas.vlan_range_start)
+        return taas_id
+
+    def create_tap_service_precommit(self, context):
+        ts = context.tap_service
+        tap_id_association = context._plugin.create_tap_id_association(
+            context._plugin_context, ts['id'])
+        context.tap_id_association = tap_id_association
+
+    def create_tap_service_postcommit(self, context):
+        """Send tap service creation RPC message to agent.
+
+        This RPC message includes taas_id that is added vlan_range_start to
+        so that taas-ovs-agent can use taas_id as VLANID.
+        """
+        # Get taas id associated with the Tap Service
+        ts = context.tap_service
+        tap_id_association = context.tap_id_association
+        taas_vlan_id = (tap_id_association['taas_id'] +
+                        cfg.CONF.taas.vlan_range_start)
+        port = self.service_plugin._get_port_details(context._plugin_context,
+                                                     ts['port_id'])
+        host = port['binding:host_id']
+
+        if taas_vlan_id > cfg.CONF.taas.vlan_range_end:
+            raise taas_ex.TapServiceLimitReached()
+
+        rpc_msg = {'tap_service': ts,
+                   'taas_id': taas_vlan_id,
+                   'port': port}
+
+        self.agent_rpc.create_tap_service(context._plugin_context,
+                                          rpc_msg, host)
+        return
+
+    def delete_tap_service_precommit(self, context):
+        pass
+
+    def delete_tap_service_postcommit(self, context):
+        """Send tap service deletion RPC message to agent.
+
+        This RPC message includes taas_id that is added vlan_range_start to
+        so that taas-ovs-agent can use taas_id as VLANID.
+        """
+        ts = context.tap_service
+        tap_id_association = context.tap_id_association
+        taas_vlan_id = (tap_id_association['taas_id'] +
+                        cfg.CONF.taas.vlan_range_start)
+        port = self.service_plugin._get_port_details(context._plugin_context,
+                                                     ts['port_id'])
+        host = port['binding:host_id']
+
+        rpc_msg = {'tap_service': ts,
+                   'taas_id': taas_vlan_id,
+                   'port': port}
+
+        self.agent_rpc.delete_tap_service(context._plugin_context,
+                                          rpc_msg, host)
+        return
+
+    def create_tap_flow_precommit(self, context):
+        pass
+
+    def create_tap_flow_postcommit(self, context):
+        """Send tap flow creation RPC message to agent."""
+        tf = context.tap_flow
+        taas_id = self._get_taas_id(context._plugin_context, tf)
+        # Extract the host where the source port is located
+        port = self.service_plugin._get_port_details(context._plugin_context,
+                                                     tf['source_port'])
+        host = port['binding:host_id']
+        port_mac = port['mac_address']
+        # Send RPC message to both the source port host and
+        # tap service(destination) port host
+        rpc_msg = {'tap_flow': tf,
+                   'port_mac': port_mac,
+                   'taas_id': taas_id,
+                   'port': port}
+
+        self.agent_rpc.create_tap_flow(context._plugin_context, rpc_msg, host)
+        return
+
+    def delete_tap_flow_precommit(self, context):
+        pass
+
+    def delete_tap_flow_postcommit(self, context):
+        """Send tap flow deletion RPC message to agent."""
+        tf = context.tap_flow
+        taas_id = self._get_taas_id(context._plugin_context, tf)
+        # Extract the host where the source port is located
+        port = self.service_plugin._get_port_details(context._plugin_context,
+                                                     tf['source_port'])
+        host = port['binding:host_id']
+        port_mac = port['mac_address']
+        # Send RPC message to both the source port host and
+        # tap service(destination) port host
+        rpc_msg = {'tap_flow': tf,
+                   'port_mac': port_mac,
+                   'taas_id': taas_id,
+                   'port': port}
+
+        self.agent_rpc.delete_tap_flow(context._plugin_context, rpc_msg, host)
+        return

neutron_taas/services/taas/taas_plugin.py

@@ -1,3 +1,4 @@
+# Copyright (C) 2016 Midokura SARL.
 # Copyright (C) 2015 Ericsson AB
 # Copyright (c) 2015 Gigamon
 #
@@ -13,76 +14,27 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+from neutron.common import exceptions as n_exc
+from neutron.db import servicetype_db as st_db
+from neutron.services import provider_configuration as pconf
+from neutron.services import service_base
 
-from oslo_config import cfg
+from neutron_taas.common import constants
-import oslo_messaging as messaging
-
-from neutron.common import rpc as n_rpc
-from neutron_taas.common import topics
 from neutron_taas.db import taas_db
 from neutron_taas.extensions import taas as taas_ex
+from neutron_taas.services.taas.service_drivers import (service_driver_context
+                                                        as sd_context)
+
 from oslo_log import log as logging
+from oslo_utils import excutils
 
 LOG = logging.getLogger(__name__)
 
 
-class TaasCallbacks(object):
+def add_provider_configuration(type_manager, service_type):
-    """Currently there are no callbacks to the Taas Plugin."""
+    type_manager.add_provider_configuration(
-
+        service_type,
-    def __init__(self, plugin):
+        pconf.ProviderConfiguration('neutron_taas'))
-        super(TaasCallbacks, self).__init__()
-        self.plugin = plugin
-        return
-
-
-class TaasAgentApi(object):
-    """RPC calls to agent APIs"""
-
-    def __init__(self, topic, host):
-        self.host = host
-        target = messaging.Target(topic=topic, version='1.0')
-        self.client = n_rpc.get_client(target)
-        return
-
-    def create_tap_service(self, context, tap_service, host):
-        LOG.debug("In RPC Call for Create Tap Service: Host=%s, MSG=%s" %
-                  (host, tap_service))
-
-        cctxt = self.client.prepare(fanout=True)
-        cctxt.cast(context, 'create_tap_service', tap_service=tap_service,
-                   host=host)
-
-        return
-
-    def create_tap_flow(self, context, tap_flow_msg, host):
-        LOG.debug("In RPC Call for Create Tap Flow: Host=%s, MSG=%s" %
-                  (host, tap_flow_msg))
-
-        cctxt = self.client.prepare(fanout=True)
-        cctxt.cast(context, 'create_tap_flow', tap_flow_msg=tap_flow_msg,
-                   host=host)
-
-        return
-
-    def delete_tap_service(self, context, tap_service, host):
-        LOG.debug("In RPC Call for Delete Tap Service: Host=%s, MSG=%s" %
-                  (host, tap_service))
-
-        cctxt = self.client.prepare(fanout=True)
-        cctxt.cast(context, 'delete_tap_service', tap_service=tap_service,
-                   host=host)
-
-        return
-
-    def delete_tap_flow(self, context, tap_flow_msg, host):
-        LOG.debug("In RPC Call for Delete Tap Flow: Host=%s, MSG=%s" %
-                  (host, tap_flow_msg))
-
-        cctxt = self.client.prepare(fanout=True)
-        cctxt.cast(context, 'delete_tap_flow', tap_flow_msg=tap_flow_msg,
-                   host=host)
-
-        return
 
 
 class TaasPlugin(taas_db.Tass_db_Mixin):
@@ -93,20 +45,24 @@ class TaasPlugin(taas_db.Tass_db_Mixin):
     def __init__(self):
 
         LOG.debug("TAAS PLUGIN INITIALIZED")
-        self.endpoints = [TaasCallbacks(self)]
+        self.service_type_manager = st_db.ServiceTypeManager.get_instance()
-
+        add_provider_configuration(self.service_type_manager, constants.TAAS)
-        self.conn = n_rpc.create_connection()
+        self._load_drivers()
-        self.conn.create_consumer(
+        self.driver = self._get_driver_for_provider(self.default_provider)
-            topics.TAAS_PLUGIN, self.endpoints, fanout=False)
-        self.conn.consume_in_threads()
-
-        self.agent_rpc = TaasAgentApi(
-            topics.TAAS_AGENT,
-            cfg.CONF.host
-        )
 
         return
 
+    def _load_drivers(self):
+        """Loads plugin-drivers specified in configuration."""
+        self.drivers, self.default_provider = service_base.load_drivers(
+            'TAAS', self)
+
+    def _get_driver_for_provider(self, provider):
+        if provider in self.drivers:
+            return self.drivers[provider]
+        raise n_exc.Invalid("Error retrieving driver for provider %s" %
+                            provider)
+
     def create_tap_service(self, context, tap_service):
         LOG.debug("create_tap_service() called")
 
@@ -130,34 +86,25 @@ class TaasPlugin(taas_db.Tass_db_Mixin):
             LOG.debug("Host could not be found, Port Binding disbaled!")
 
         # Create tap service in the db model
-        ts = super(TaasPlugin, self).create_tap_service(context, tap_service)
+        with context.session.begin(subtransactions=True):
-        # Get taas id associated with the Tap Service
+            ts = super(TaasPlugin, self).create_tap_service(context,
-        tap_id_association = self.get_tap_id_association(
+                                                            tap_service)
-            context,
+            driver_context = sd_context.TapServiceContext(self, context, ts)
-            tap_service_id=ts['id'])
+            self.driver.create_tap_service_precommit(driver_context)
 
-        taas_vlan_id = (tap_id_association['taas_id'] +
+        try:
-                        cfg.CONF.taas.vlan_range_start)
+            self.driver.create_tap_service_postcommit(driver_context)
-
+        except Exception:
-        if taas_vlan_id > cfg.CONF.taas.vlan_range_end:
+            with excutils.save_and_reraise_exception():
-            raise taas_ex.TapServiceLimitReached()
+                LOG.error("Failed to create tap service on driver,"
-
+                          "deleting tap_service %s", ts['id'])
-        rpc_msg = {'tap_service': ts, 'taas_id': taas_vlan_id, 'port': port}
+                super(TaasPlugin, self).delete_tap_service(context, ts['id'])
-
-        self.agent_rpc.create_tap_service(context, rpc_msg, host)
 
         return ts
 
     def delete_tap_service(self, context, id):
         LOG.debug("delete_tap_service() called")
 
-        # Get taas id associated with the Tap Service
-        tap_id_association = self.get_tap_id_association(
-            context,
-            tap_service_id=id)
-
-        ts = self.get_tap_service(context, id)
-
         # Get all the tap Flows that are associated with the Tap service
         # and delete them as well
         t_f_collection = self.get_tap_flows(
@@ -167,25 +114,18 @@ class TaasPlugin(taas_db.Tass_db_Mixin):
         for t_f in t_f_collection:
             self.delete_tap_flow(context, t_f['id'])
 
-        # Get the port and the host that it is on
+        with context.session.begin(subtransactions=True):
-        port_id = ts['port_id']
+            ts = self.get_tap_service(context, id)
-
+            driver_context = sd_context.TapServiceContext(self, context, ts)
-        port = self._get_port_details(context, port_id)
-
-        host = port['binding:host_id']
-
             super(TaasPlugin, self).delete_tap_service(context, id)
+            self.driver.delete_tap_service_precommit(driver_context)
 
-        taas_vlan_id = (tap_id_association['taas_id'] +
+        try:
-                        cfg.CONF.taas.vlan_range_start)
+            self.driver.delete_tap_service_postcommit(driver_context)
-
+        except Exception:
-        rpc_msg = {'tap_service': ts,
+            with excutils.save_and_reraise_exception():
-                   'taas_id': taas_vlan_id,
+                LOG.error("Failed to delete tap service on driver. "
-                   'port': port}
+                          "tap_sevice: %s", id)
-
-        self.agent_rpc.delete_tap_service(context, rpc_msg, host)
-
-        return ts
 
     def create_tap_flow(self, context, tap_flow):
         LOG.debug("create_tap_flow() called")
@@ -199,56 +139,38 @@ class TaasPlugin(taas_db.Tass_db_Mixin):
         ts = self.get_tap_service(context, t_f['tap_service_id'])
         ts_tenant_id = ts['tenant_id']
 
-        taas_id = (self.get_tap_id_association(
-            context,
-            tap_service_id=ts['id'])['taas_id'] +
-            cfg.CONF.taas.vlan_range_start)
-
         if tenant_id != ts_tenant_id:
             raise taas_ex.TapServiceNotBelongToTenant()
 
-        # Extract the host where the source port is located
-        port = self._get_port_details(context, t_f['source_port'])
-        host = port['binding:host_id']
-        port_mac = port['mac_address']
-
         # create tap flow in the db model
+        with context.session.begin(subtransactions=True):
             tf = super(TaasPlugin, self).create_tap_flow(context, tap_flow)
+            driver_context = sd_context.TapFlowContext(self, context, tf)
+            self.driver.create_tap_flow_precommit(driver_context)
 
-        # Send RPC message to both the source port host and
+        try:
-        # tap service(destination) port host
+            self.driver.create_tap_flow_postcommit(driver_context)
-        rpc_msg = {'tap_flow': tf,
+        except Exception:
-                   'port_mac': port_mac,
+            with excutils.save_and_reraise_exception():
-                   'taas_id': taas_id,
+                LOG.error("Failed to create tap flow on driver,"
-                   'port': port}
+                          "deleting tap_flow %s", tf['id'])
-
+                super(TaasPlugin, self).delete_tap_flow(context, tf['id'])
-        self.agent_rpc.create_tap_flow(context, rpc_msg, host)
 
         return tf
 
     def delete_tap_flow(self, context, id):
         LOG.debug("delete_tap_flow() called")
 
+        with context.session.begin(subtransactions=True):
             tf = self.get_tap_flow(context, id)
-
+            driver_context = sd_context.TapFlowContext(self, context, tf)
-        taas_id = (self.get_tap_id_association(
-            context,
-            tf['tap_service_id'])['taas_id'] +
-            cfg.CONF.taas.vlan_range_start)
-
-        port = self._get_port_details(context, tf['source_port'])
-        host = port['binding:host_id']
-        port_mac = port['mac_address']
-
             super(TaasPlugin, self).delete_tap_flow(context, id)
+            self.driver.delete_tap_flow_precommit(driver_context)
 
-        # Send RPC message to both the source port host and
+        try:
-        # tap service(destination) port host
+            self.driver.delete_tap_flow_postcommit(driver_context)
-        rpc_msg = {'tap_flow': tf,
+        except Exception:
-                   'port_mac': port_mac,
+            with excutils.save_and_reraise_exception():
-                   'taas_id': taas_id,
+                with excutils.save_and_reraise_exception():
-                   'port': port}
+                    LOG.error("Failed to delete tap flow on driver. "
-
+                              "tap_flow: %s", id)
-        self.agent_rpc.delete_tap_flow(context, rpc_msg, host)
-
-        return tf

neutron_taas/tests/unit/services/taas/test_taas_plugin.py

@@ -18,7 +18,6 @@ import contextlib
 import mock
 import testtools
 
-from oslo_config import cfg
 from oslo_utils import uuidutils
 
 import neutron.common.rpc as n_rpc
@@ -28,15 +27,28 @@ from neutron.tests.unit import testlib_api
 
 import neutron_taas.db.taas_db  # noqa
 import neutron_taas.extensions.taas as taas_ext
+from neutron_taas.services.taas.service_drivers import taas_agent_api
 from neutron_taas.services.taas import taas_plugin
 
 
+class DummyError(Exception):
+    pass
+
+
 class TestTaasPlugin(testlib_api.SqlTestCase):
     def setUp(self):
         super(TestTaasPlugin, self).setUp()
         mock.patch.object(n_rpc, 'create_connection', auto_spec=True).start()
-        mock.patch.object(taas_plugin, 'TaasCallbacks', auto_spec=True).start()
+        mock.patch.object(taas_agent_api,
-        mock.patch.object(taas_plugin, 'TaasAgentApi', auto_spec=True).start()
+                          'TaasCallbacks', auto_spec=True).start()
+        mock.patch.object(taas_agent_api,
+                          'TaasAgentApi', auto_spec=True).start()
+        self.driver = mock.MagicMock()
+        mock.patch('neutron.services.service_base.load_drivers',
+                   return_value=({'dummy_provider': self.driver},
+                                 'dummy_provider')).start()
+        mock.patch('neutron.db.servicetype_db.ServiceTypeManager.get_instance',
+                   return_value=mock.MagicMock()).start()
         self._plugin = taas_plugin.TaasPlugin()
         self._context = context.get_admin_context()
 
@@ -73,15 +85,17 @@ class TestTaasPlugin(testlib_api.SqlTestCase):
                                return_value=self._port_details):
             yield self._plugin.create_tap_service(self._context, req)
         self._tap_service['id'] = mock.ANY
-        expected_msg = {
+
-            'tap_service': self._tap_service,
+        self.driver.assert_has_calls([
-            'taas_id': mock.ANY,
+            mock.call.create_tap_service_precommit(mock.ANY),
-            'port': self._port_details,
+            mock.call.create_tap_service_postcommit(mock.ANY),
-        }
-        self._plugin.agent_rpc.assert_has_calls([
-            mock.call.create_tap_service(self._context, expected_msg,
-                                         self._host_id),
         ])
+        pre_args = self.driver.create_tap_service_precommit.call_args[0][0]
+        self.assertEqual(self._context, pre_args._plugin_context)
+        self.assertEqual(self._tap_service, pre_args.tap_service)
+        post_args = self.driver.create_tap_service_postcommit.call_args[0][0]
+        self.assertEqual(self._context, post_args._plugin_context)
+        self.assertEqual(self._tap_service, post_args.tap_service)
 
     @contextlib.contextmanager
     def tap_flow(self, tap_service, tenant_id=None):
@@ -96,16 +110,17 @@ class TestTaasPlugin(testlib_api.SqlTestCase):
             yield self._plugin.create_tap_flow(self._context, req)
         self._tap_flow['id'] = mock.ANY
         self._tap_service['id'] = mock.ANY
-        expected_msg = {
+
-            'tap_flow': self._tap_flow,
+        self.driver.assert_has_calls([
-            'port_mac': self._port_details['mac_address'],
+            mock.call.create_tap_flow_precommit(mock.ANY),
-            'taas_id': mock.ANY,
+            mock.call.create_tap_flow_postcommit(mock.ANY),
-            'port': self._port_details,
-        }
-        self._plugin.agent_rpc.assert_has_calls([
-            mock.call.create_tap_flow(self._context, expected_msg,
-                                      self._host_id),
         ])
+        pre_args = self.driver.create_tap_flow_precommit.call_args[0][0]
+        self.assertEqual(self._context, pre_args._plugin_context)
+        self.assertEqual(self._tap_flow, pre_args.tap_flow)
+        post_args = self.driver.create_tap_flow_postcommit.call_args[0][0]
+        self.assertEqual(self._context, post_args._plugin_context)
+        self.assertEqual(self._tap_flow, post_args.tap_flow)
 
     def test_create_tap_service(self):
         with self.tap_service():
@@ -116,57 +131,71 @@ class TestTaasPlugin(testlib_api.SqlTestCase):
         with testtools.ExpectedException(taas_ext.PortDoesNotBelongToTenant), \
             self.tap_service():
             pass
-        self.assertEqual([], self._plugin.agent_rpc.mock_calls)
+        self.assertEqual([], self.driver.mock_calls)
 
     def test_create_tap_service_reach_limit(self):
-        cfg.CONF.set_override('vlan_range_end', 3900, 'taas')  # same as start
+        # TODO(Yoichiro):Need to move this test to taas_rpc test
-        with testtools.ExpectedException(taas_ext.TapServiceLimitReached), \
-            self.tap_service():
         pass
-        self.assertEqual([], self._plugin.agent_rpc.mock_calls)
+
+    def test_create_tap_service_failed_on_service_driver(self):
+        attr = {'create_tap_service_postcommit.side_effect': DummyError}
+        self.driver.configure_mock(**attr)
+        with testtools.ExpectedException(DummyError):
+            req = {
+                'tap_service': self._tap_service,
+            }
+            with mock.patch.object(self._plugin, '_get_port_details',
+                                   return_value=self._port_details):
+                self._plugin.create_tap_service(self._context, req)
 
     def test_delete_tap_service(self):
         with self.tap_service() as ts:
             self._plugin.delete_tap_service(self._context, ts['id'])
-        expected_msg = {
+        self.driver.assert_has_calls([
-            'tap_service': self._tap_service,
+            mock.call.delete_tap_service_precommit(mock.ANY),
-            'taas_id': mock.ANY,
+            mock.call.delete_tap_service_postcommit(mock.ANY),
-            'port': self._port_details,
-        }
-        self._plugin.agent_rpc.assert_has_calls([
-            mock.call.delete_tap_service(self._context, expected_msg,
-                                         self._host_id),
         ])
+        pre_args = self.driver.delete_tap_service_precommit.call_args[0][0]
+        self.assertEqual(self._context, pre_args._plugin_context)
+        self.assertEqual(self._tap_service, pre_args.tap_service)
+        post_args = self.driver.delete_tap_service_postcommit.call_args[0][0]
+        self.assertEqual(self._context, post_args._plugin_context)
+        self.assertEqual(self._tap_service, post_args.tap_service)
 
     def test_delete_tap_service_with_flow(self):
         with self.tap_service() as ts, \
-            self.tap_flow(tap_service=ts['id']) as tf:
+            self.tap_flow(tap_service=ts['id']):
             self._plugin.delete_tap_service(self._context, ts['id'])
-        expected_msg = {
+        self.driver.assert_has_calls([
-            'tap_service': self._tap_service,
+            mock.call.delete_tap_flow_precommit(mock.ANY),
-            'taas_id': mock.ANY,
+            mock.call.delete_tap_flow_postcommit(mock.ANY),
-            'port': self._port_details,
+            mock.call.delete_tap_service_precommit(mock.ANY),
-        }
+            mock.call.delete_tap_service_postcommit(mock.ANY),
-        self._plugin.agent_rpc.assert_has_calls([
-            mock.call.delete_tap_service(self._context, expected_msg,
-                                         self._host_id),
-        ])
-        self._tap_flow['id'] = tf['id']
-        expected_msg = {
-            'tap_flow': self._tap_flow,
-            'taas_id': mock.ANY,
-            'port_mac': self._port_details['mac_address'],
-            'port': self._port_details,
-        }
-        self._plugin.agent_rpc.assert_has_calls([
-            mock.call.delete_tap_flow(self._context, expected_msg,
-                                      self._host_id),
         ])
+        pre_args = self.driver.delete_tap_flow_precommit.call_args[0][0]
+        self.assertEqual(self._context, pre_args._plugin_context)
+        self.assertEqual(self._tap_flow, pre_args.tap_flow)
+        post_args = self.driver.delete_tap_flow_postcommit.call_args[0][0]
+        self.assertEqual(self._context, post_args._plugin_context)
+        self.assertEqual(self._tap_flow, post_args.tap_flow)
+        pre_args = self.driver.delete_tap_service_precommit.call_args[0][0]
+        self.assertEqual(self._context, pre_args._plugin_context)
+        self.assertEqual(self._tap_service, pre_args.tap_service)
+        post_args = self.driver.delete_tap_service_postcommit.call_args[0][0]
+        self.assertEqual(self._context, post_args._plugin_context)
+        self.assertEqual(self._tap_service, post_args.tap_service)
 
     def test_delete_tap_service_non_existent(self):
         with testtools.ExpectedException(taas_ext.TapServiceNotFound):
             self._plugin.delete_tap_service(self._context, 'non-existent')
 
+    def test_delete_tap_service_failed_on_service_driver(self):
+        attr = {'delete_tap_service_postcommit.side_effect': DummyError}
+        self.driver.configure_mock(**attr)
+        with self.tap_service() as ts:
+            with testtools.ExpectedException(DummyError):
+                self._plugin.delete_tap_service(self._context, ts['id'])
+
     def test_create_tap_flow(self):
         with self.tap_service() as ts, self.tap_flow(tap_service=ts['id']):
             pass
@@ -177,18 +206,39 @@ class TestTaasPlugin(testlib_api.SqlTestCase):
             self.tap_flow(tap_service=ts['id'], tenant_id='other-tenant'):
             pass
 
+    def test_create_tap_flow_failed_on_service_driver(self):
+        with self.tap_service() as ts:
+            attr = {'create_tap_flow_postcommit.side_effect': DummyError}
+            self.driver.configure_mock(**attr)
+            with testtools.ExpectedException(DummyError):
+                self._tap_flow['tap_service_id'] = ts['id']
+                req = {
+                    'tap_flow': self._tap_flow,
+                }
+                with mock.patch.object(self._plugin, '_get_port_details',
+                                       return_value=self._port_details):
+                    self._plugin.create_tap_flow(self._context, req)
+
     def test_delete_tap_flow(self):
         with self.tap_service() as ts, \
             self.tap_flow(tap_service=ts['id']) as tf:
             self._plugin.delete_tap_flow(self._context, tf['id'])
         self._tap_flow['id'] = tf['id']
-        expected_msg = {
+        self.driver.assert_has_calls([
-            'tap_flow': self._tap_flow,
+            mock.call.delete_tap_flow_precommit(mock.ANY),
-            'taas_id': mock.ANY,
+            mock.call.delete_tap_flow_postcommit(mock.ANY),
-            'port_mac': self._port_details['mac_address'],
-            'port': self._port_details,
-        }
-        self._plugin.agent_rpc.assert_has_calls([
-            mock.call.delete_tap_flow(self._context, expected_msg,
-                                      self._host_id),
         ])
+        pre_args = self.driver.delete_tap_flow_precommit.call_args[0][0]
+        self.assertEqual(self._context, pre_args._plugin_context)
+        self.assertEqual(self._tap_flow, pre_args.tap_flow)
+        post_args = self.driver.delete_tap_flow_postcommit.call_args[0][0]
+        self.assertEqual(self._context, post_args._plugin_context)
+        self.assertEqual(self._tap_flow, post_args.tap_flow)
+
+    def test_delete_tap_flow_failed_on_service_driver(self):
+        with self.tap_service() as ts, \
+            self.tap_flow(tap_service=ts['id']) as tf:
+            attr = {'delete_tap_flow_postcommit.side_effect': DummyError}
+            self.driver.configure_mock(**attr)
+            with testtools.ExpectedException(DummyError):
+                self._plugin.delete_tap_flow(self._context, tf['id'])